﻿#include "ChatClient.h"
#include "XyNet/XyNet.h"

XYNET_IMPLEMENT_BASE_SINGLETON(XyNetTest::TestClientManager);

namespace XyNetTest
{
	ChatClient::ChatClient(const XyNet::Id& id, const string& queries)
		: _clientId(id), _showLog(XyNet::Logger::getDefault()->getWriteToConsole())
	{
		auto config = make_shared<XyNet::RpcClient::Config>();
		config->setQueries(queries);
		config->setMessageTimeout(9999);	// 避免测试时超时异常
		//config->setMaxMessagePacketSize(999999999);
		//config->setMaxMessageCountPerSecond(5000);
		//config->setMaxDownloadSpeed(2000);
		//config->setMaxUploadSpeed(2000);
		//config->setEncryptionLevel(3);

		//config->setHeartBeatInterval(1);
		//config->setMaxDownloadSpeed(1500);
		//config->setMaxUploadSpeed(2000);
		//config->setMaxNumberOfMessagesPerSecond(1);
		//config->setMaxMessagePacketSize(1);
		//config->setMaxNumberOfMessagesPerSecond(1);

		_client = make_shared<XyNet::RpcClient>(config);
		_client->getServiceManager()->addService(
			make_shared<ClientUserServiceImpl>(id, _client.get()));
	}

	shared_ptr<XyNet::RpcClient> ChatClient::getClient()
	{
		return _client;
	}

	void ChatClient::addVerifyMessage(const string& message)
	{
		unique_lock<mutex> lock(_messageMutex);
		_verifyMessages.emplace_back(message);
		lock.unlock();
	}

	void ChatClient::addMessage(const string& message)
	{
		unique_lock<mutex> lock(_messageMutex);
		_messages.emplace_back(message);
		lock.unlock();

		if (_showLog)
			XyNet::Logger::getDefault()->log(message);
	}

	bool ChatClient::verify()
	{
		// 等待异步操作完成
		auto socketChannel = 
			XYNET_POINTER_DOWNCAST<XyNet::RpcSocketChannel>(_client->getChannel());
		while (_messages.size() < _verifyMessages.size() ||
			socketChannel->getCallbackCount() > 0)
		{
			this_thread::sleep_for(chrono::milliseconds(10));
		}

		unique_lock<mutex> lock(_messageMutex);
		if (_messages.size() > 0)
		{
			XYNET_CHECK(_messages.size() == _verifyMessages.size());
			_verifyMessages.sort();
			_messages.sort();

			decltype(_messages) result;	// 尺寸为0，一旦有差集就会出错
			std::set_difference(_verifyMessages.begin(), _verifyMessages.end(),
				_messages.begin(), _messages.end(), result.begin());

			_verifyMessages.clear();
			_messages.clear();
		}
		lock.unlock();
		return true;
	}

	void ChatClient::login(const XyNet::Id& userId)
	{
		boost::format format("<客户端%d, 用户%d>: 登陆用户%d。");
		format % _clientId % _client->getUserId() % userId;
		addVerifyMessage(format.str());
		addMessage(format.str());

		ServerUserService::LoginMethod method(_client->getChannel());
		method.getRequest()->set_user_id(userId);
		method.call([this](ServerUserService::LoginMethod::Argument* argument)
		{
			auto controller = argument->getController();
			auto userId = argument->getRequest()->user_id();
			controller->setUserId(userId);
			controller->getSocketChannelServiceManager()->addService(
				make_shared<ClientChatServiceImpl>(_clientId));

			boost::format format("<客户端%d, 用户%d>: 登陆用户%d成功。");
			format % _clientId % _client->getUserId() % userId;
			addMessage(format.str());
		}, [this](ServerUserService::LoginMethod::Argument* argument)
		{
			boost::format format("<客户端%d, 用户%d>: %s。");
			auto error = argument->getController()->getError();
			auto errorMessage = 
				error->getDetail().empty() ? error->getBrief() : error->getDetail();
			format % _clientId % _client->getUserId() % errorMessage;
			addMessage(format.str());
		});
	}

	void ChatClient::logout()
	{
		auto userId = _client->getUserId();

		boost::format format("<客户端%d, 用户%d>: 登出用户%d。");
		format % _clientId % _client->getUserId() % userId;
		addVerifyMessage(format.str());
		addMessage(format.str());

		ServerUserService::LogoutMethod method(_client->getChannel());
		method.call([this](ServerUserService::LogoutMethod::Argument* argument)
		{
			auto controller = argument->getController();
			auto userId = controller->getUserId();
			controller->setUserId(0);
			controller->getSocketChannelServiceManager()->clear();

			boost::format format("<客户端%d, 用户%d>: 登出用户%d成功。");
			format % _clientId % _client->getUserId() % userId;
			addMessage(format.str());
		},	[this](ServerUserService::LogoutMethod::Argument* argument)
		{
			boost::format format("<客户端%d, 用户%d>: %s。");
			auto error = argument->getController()->getError();
			auto errorMessage =
				error->getDetail().empty() ? error->getBrief() : error->getDetail();
			format % _clientId % _client->getUserId() % errorMessage;
			addMessage(format.str());
		});
	}

	void ChatClient::publicChat(const string& message)
	{
		boost::format verifyFormat("<客户端%d, 用户%d>: 发送公聊消息%s。");
		verifyFormat % _clientId % _client->getUserId() % message;
		addVerifyMessage(verifyFormat.str());
		addMessage(verifyFormat.str());

		ServerChatService::PublicSayMethod method(_client->getChannel());
		method.getRequest()->set_message(message);
		method.call([](ServerChatService::PublicSayMethod::Argument*) {
		}, [this](ServerChatService::PublicSayMethod::Argument* argument)
		{
			boost::format format("<客户端%d, 用户%d>: %s。");
			auto error = argument->getController()->getError();
			auto errorMessage =
				error->getDetail().empty() ? error->getBrief() : error->getDetail();
			format % _clientId % _client->getUserId() % errorMessage;
			addMessage(format.str());
		});
	}

	void ChatClient::privateChat(const XyNet::Id& otherId, const string& message)
	{
		boost::format verifyFormat("<客户端%d, 用户%d>: 向用户%d发送私聊消息: %s。");
		verifyFormat % _clientId % _client->getUserId() % otherId % message;
		addVerifyMessage(verifyFormat.str());
		addMessage(verifyFormat.str());

		ServerChatService::PrivateSayMethod method(_client->getChannel());
		method.getRequest()->set_other_user_id(otherId);
		method.getRequest()->set_message(message);
		method.call([this](ServerChatService::PrivateSayMethod::Argument* argument)
		{
			auto request = argument->getRequest();
			boost::format format("<客户端%d, 用户%d>: 成功向用户%d发送私聊消息: %s。");
			format % _clientId % _client->getUserId() % request->other_user_id() % request->message();
			addMessage(format.str());
		},[this](ServerChatService::PrivateSayMethod::Argument* argument)
		{
			boost::format format("<客户端%d, 用户%d>: %s。");
			auto error = argument->getController()->getError();
			auto errorMessage =
				error->getDetail().empty() ? error->getBrief() : error->getDetail();
			format % _clientId % _client->getUserId() % errorMessage;
			addMessage(format.str());
		});
	}

	//////////////////////////////////////////////////////////////////////////

	TestClientManager::~TestClientManager()
	{
		clearTestClients();
	}

	void TestClientManager::test(const string& query, int clientNumber)
	{
		clearTestClients();
		createTestClients(query, clientNumber);

		// 显示错误日志
		XyNet::Logger::getDefault()->setWriteToConsole(true);

		if (clientNumber < 3)
		{
			XyNet::Logger::getDefault()->logWarning("测试客户端数量小于3。");
			return;
		}

		// 将客户端划分为3部分，0~0.5是正常登陆用户，0.5~0.75是登陆后被挤下线的用户，0.75到1是未登陆用户
		XyNet::Id kickedClientBeginId = clientNumber / 2 + 1;
		XyNet::Id notLoginClientBeginId = kickedClientBeginId + (clientNumber + 1) / 4;

		// 登陆要被踢的用户
		for (XyNet::Id clientId = kickedClientBeginId; 
			clientId < notLoginClientBeginId; ++clientId)
		{
			XyNet::Id userId = clientId - kickedClientBeginId + 1;
			_idTestClientMap[clientId]->login(userId);
			// 所有客户端都能收到消息
			for (const auto& pair : _idTestClientMap)
			{
				auto testClient = pair.second;
				boost::format format("<客户端%d>: 发现用户%d已登录。");
				format % pair.first % userId;
				pair.second->addVerifyMessage(format.str());
			}

			boost::format format("<客户端%d, 用户%d>: 登陆用户%d成功。");
			format % clientId % userId % userId;
			_idTestClientMap[clientId]->addVerifyMessage(format.str());
		}
		verify("验证登陆要被踢的用户");	// 等待登陆完成，避免正常登陆用户被要被踢的用户踢

		// 登录正常用户
		for (XyNet::Id clientId = 1; clientId < kickedClientBeginId; ++clientId)
		{
			XyNet::Id userId = clientId;
			_idTestClientMap[clientId]->login(userId);
			// 所有客户端都能收到消息
			for (const auto& pair : _idTestClientMap)
			{
				auto testClient = pair.second;
				boost::format format("<客户端%d>: 发现用户%d已登录。");
				format % pair.first % userId;
				pair.second->addVerifyMessage(format.str());
			}

			boost::format format("<客户端%d, 用户%d>: 登陆用户%d成功。");
			format % clientId % userId % userId;
			_idTestClientMap[clientId]->addVerifyMessage(format.str());
		}

		for (XyNet::Id clientId = kickedClientBeginId; 
			clientId < notLoginClientBeginId; ++clientId)
		{
			XyNet::Id userId = clientId - kickedClientBeginId + 1;
			boost::format format2("<客户端%d, 用户%d>: 被踢出。");
			_idTestClientMap[clientId]->addVerifyMessage((format2 % clientId % userId).str());
		}
		verify("验证登录正常用户");

		// 再次登录正常用户
		for (XyNet::Id clientId = 1; clientId < kickedClientBeginId; ++clientId)
		{
			XyNet::Id userId = clientId;
			_idTestClientMap[clientId]->login(userId);

			boost::format format("<客户端%d, 用户%d>: 已经是登陆状态。");
			format % clientId % userId;
			_idTestClientMap[clientId]->addVerifyMessage(format.str());
		}
		verify("验证再次登录正常用户");

		// 公聊
		for (XyNet::Id clientId = 1; clientId <= clientNumber; ++clientId)
		{
			boost::format messageFormat("用户%s公聊");
			auto message = (messageFormat % clientId).str();

			_idTestClientMap[clientId]->publicChat(message);
		}
		for (XyNet::Id loginClientId = 1; loginClientId < kickedClientBeginId; ++loginClientId)
		{
			for (XyNet::Id otherClientId = 1; otherClientId < kickedClientBeginId; ++otherClientId)
			{
				boost::format format("<客户端%d, 用户%d>: 收到用户%d公聊消息: %s。");
				format % loginClientId % 
					_idTestClientMap[loginClientId]->getClient()->getUserId() %
					otherClientId % ("用户" + lexical_cast<string>(otherClientId) + "公聊");
				_idTestClientMap[loginClientId]->addVerifyMessage(format.str());
			}
		}
		for (XyNet::Id noLoginClientId = kickedClientBeginId;
			noLoginClientId <= clientNumber; ++noLoginClientId)
		{
			boost::format format("<客户端%d, 用户%d>: Service not found。");
			format % noLoginClientId % 0;
			_idTestClientMap[noLoginClientId]->addVerifyMessage(format.str());
		}
		verify("验证公聊");

		// 私聊
		for (int clientId = 1; clientId < kickedClientBeginId; ++clientId)
		{
			for (int otherClientId = 1; otherClientId <= clientNumber; ++otherClientId)
			{
				boost::format messageFormat("%d私聊%d");
				messageFormat % clientId % otherClientId;
				_idTestClientMap[clientId]->privateChat(otherClientId, messageFormat.str());
			}
		}
		for (int clientId = 1; clientId < kickedClientBeginId; ++clientId)
		{
			for (XyNet::Id otherClientId = 1; otherClientId < kickedClientBeginId; 
				++otherClientId)
			{
				boost::format messageFormat("%d私聊%d");
				messageFormat % clientId % otherClientId;
				boost::format format("<客户端%d, 用户%d>: 收到用户%d私聊消息: %s。");
				format % otherClientId % 
					_idTestClientMap[otherClientId]->getClient()->getUserId() %
					clientId % messageFormat.str();
				_idTestClientMap[otherClientId]->addVerifyMessage(format.str());

				boost::format format2("<客户端%d, 用户%d>: 成功向用户%d发送私聊消息: %s。");
				format2 % clientId % clientId % otherClientId % messageFormat.str();
				_idTestClientMap[clientId]->addVerifyMessage(format2.str());
			}
			for (XyNet::Id otherClientId = kickedClientBeginId;
				otherClientId <= clientNumber; ++otherClientId)
			{
				boost::format format("<客户端%d, 用户%d>: 目标用户%d未登陆。");
				format % clientId % clientId % otherClientId;
				_idTestClientMap[clientId]->addVerifyMessage(format.str());
			}
		}
		verify("验证私聊");

		// 登出
		for (XyNet::Id clientId = 1; clientId < kickedClientBeginId; ++clientId)
		{
			XyNet::Id userId = clientId;
			_idTestClientMap[clientId]->logout();
			// 所有客户端都能收到消息
			for (const auto& pair : _idTestClientMap)
			{
				auto testClient = pair.second;
				boost::format format("<客户端%d>: 发现用户%d已登出。");
				format % pair.first % userId;
				pair.second->addVerifyMessage(format.str());
			}

			boost::format format("<客户端%d, 用户%d>: 登出用户%d成功。");
			format % clientId % 0 % userId;
			_idTestClientMap[clientId]->addVerifyMessage(format.str());
		}
		verify("验证登出");

		// 再次登出
		for (XyNet::Id clientId = 1; clientId < kickedClientBeginId; ++clientId)
		{
			_idTestClientMap[clientId]->logout();

			boost::format format("<客户端%d, 用户%d>: 已经是登出状态。");
			format % clientId % 0;
			_idTestClientMap[clientId]->addVerifyMessage(format.str());
		}
		verify("验证再次登出");

		// 退出后公聊
		for (XyNet::Id clientId = 1; clientId <= clientNumber; ++clientId)
		{
			_idTestClientMap[clientId]->publicChat(
				"用户" + lexical_cast<string>(clientId) + "公聊");
			boost::format format("<客户端%d, 用户%d>: Service not found。");
			format % clientId % 0;
			_idTestClientMap[clientId]->addVerifyMessage(format.str());
		}
		verify("验证退出后公聊");

		// 统计
		int64 totalHandleMessage = 0;
		auto averageDelay = chrono::steady_clock::duration(0);
		auto minDelay = chrono::steady_clock::duration(0);
		auto maxDelay = chrono::steady_clock::duration(0);
		int maxReadMessageCountPerSecond = 0;
		int maxWriteMessageCountPerSecond = 0;
		int maxReadBytesPerSecond = 0;
		int maxWriteBytesPerSecond = 0;

		for (const auto& pair : _idTestClientMap)
		{
			auto testClient = pair.second;
			auto channel = XYNET_POINTER_DOWNCAST<XyNet::RpcSocketChannel>(
				testClient->getClient()->getChannel());
			averageDelay += channel->getAverageDelay();
			if (totalHandleMessage == 0)
			{
				minDelay = channel->getMinDelay();
				maxDelay = channel->getMaxDelay();
			}
			else
			{
				if (channel->getMinDelay() < minDelay)
					minDelay = channel->getMinDelay();
				if (channel->getMaxDelay() > maxDelay)
					maxDelay = channel->getMaxDelay();
			}
			totalHandleMessage += channel->getTotalHandleMessage();

			if (channel->getMaxReadMessageCountPerSecond() > maxReadMessageCountPerSecond)
				maxReadMessageCountPerSecond = channel->getMaxReadMessageCountPerSecond();
			if (channel->getMaxWriteMessageCountPerSecond() > maxWriteMessageCountPerSecond)
				maxWriteMessageCountPerSecond = channel->getMaxWriteMessageCountPerSecond();

			if (channel->getMaxReadBytesPerSecond() > maxReadBytesPerSecond)
				maxReadBytesPerSecond = channel->getMaxReadBytesPerSecond();
			if (channel->getMaxWriteBytesPerSecond() > maxWriteBytesPerSecond)
				maxWriteBytesPerSecond = channel->getMaxWriteBytesPerSecond();
		}
		averageDelay /= clientNumber;

		boost::format statisticsFormat(
			"总客户端数量：%d\n"
			"总处理消息数量：%d\n"
			"平均延时：%lf\n"
			"最小延时：%lf\n"
			"最大延时：%lf\n"
			"最大读取消息数量：%d\n"
			"最大写入消息数量：%d\n"
			"最大读取速度：%d\n"
			"最大写入速度：%d"
		);
		statisticsFormat % clientNumber % totalHandleMessage % 
			chrono::duration_cast<chrono::duration<double>>(averageDelay).count() %
			chrono::duration_cast<chrono::duration<double>>(minDelay).count() % 
			chrono::duration_cast<chrono::duration<double>>(maxDelay).count() %
			maxReadMessageCountPerSecond % maxWriteMessageCountPerSecond %
			maxReadBytesPerSecond % maxWriteBytesPerSecond;
		auto* logger = XyNet::Logger::getDefault();
		bool savedSetting = logger->getWriteToConsole();
		logger->setWriteToConsole(true);
		logger->log(statisticsFormat.str());
		logger->log("测试成功。");
		logger->setWriteToConsole(savedSetting);
	}

	shared_ptr<ChatClient> TestClientManager::getTestClient(const XyNet::Id& clientId)
	{
		return XyNet::SetMapHelper::getValue(&_idTestClientMap, clientId);
	}

	void TestClientManager::createTestClients(const string& query, int clientNumber)
	{
		bool ready = false;
		int deltaIndex = 10000;
		int beginIndex = 0;
		int endIndex = 0;
		while (endIndex < clientNumber)
		{
			endIndex = beginIndex + deltaIndex;
			if (endIndex > clientNumber)
				endIndex = clientNumber;

			// 创建客户端
			for (int i = beginIndex; i < endIndex; ++i)
			{
				XyNet::Id clientId = _idGenerator();
				auto client = make_shared<ChatClient>(clientId, query);
				XyNet::SetMapHelper::insert(&_idTestClientMap, clientId, client);
			}

			// 等待客户端连接完成
			for (int i = beginIndex; i < endIndex; ++i)
			{
				while (!_idTestClientMap[i + 1]->getClient()->isConnectComplete())
				{
					this_thread::sleep_for(chrono::milliseconds(10));
				}
			}

			beginIndex = endIndex;
		}
		
		// 等待服务端完成
		this_thread::sleep_for(chrono::milliseconds(100));
	}

	void TestClientManager::clearTestClients()
	{
		if (!_idTestClientMap.empty())
		{
			_idTestClientMap.clear();
			_idGenerator.reset();
		}
	}

	void TestClientManager::verify(const string& desc)
	{
		XyNet::Logger::getDefault()->log(desc);
		for (const auto& pair : _idTestClientMap)
		{
			XYNET_CHECK(pair.second->verify());
		}
	}

	//////////////////////////////////////////////////////////////////////////

	ClientUserServiceImpl::ClientUserServiceImpl(const XyNet::Id& id, XyNet::RpcClient* client)
		: _clientId(id), _client(client) {}

	void ClientUserServiceImpl::onElsewhereLogin(ElsewhereLoginMethod::Argument* argument)
	{
		auto controller = argument->getController();

		//ClientUserService::OtherLogoutMethod otherLogoutMethod;
		//otherLogoutMethod.getRequest()->set_other_user_id(controller->getUserId());
		//auto broadcastChannel = controller->createBroadcastChannel();
		//broadcastChannel->setBroadcastAllUsers();
		//otherLogoutMethod.call(broadcastChannel);

		boost::format format("<客户端%d, 用户%d>: 被踢出。");
		format % _clientId % controller->getUserId();
		TestClientManager::getInstance()->getTestClient(_clientId)->addMessage(format.str());

		controller->setUserId(0);
	}

	void ClientUserServiceImpl::onOtherLogin(OtherLoginMethod::Argument* argument)
	{
		boost::format format("<客户端%d>: 发现用户%d已登录。");
		format % _clientId % argument->getRequest()->other_user_id();
		TestClientManager::getInstance()->getTestClient(_clientId)->addMessage(format.str());
	}

	void ClientUserServiceImpl::onOtherLogout(OtherLogoutMethod::Argument* argument)
	{
		boost::format format("<客户端%d>: 发现用户%d已登出。");
		format % _clientId % argument->getRequest()->other_user_id();
		TestClientManager::getInstance()->getTestClient(_clientId)->addMessage(format.str());
	}

	//////////////////////////////////////////////////////////////////////////

	ClientChatServiceImpl::ClientChatServiceImpl(const XyNet::Id& id) : _clientId(id)
	{
	}

	void ClientChatServiceImpl::onPushMessage(PushMessageMethod::Argument* argument)
	{
		auto controller = argument->getController();
		boost::format format("<客户端%d, 用户%d>: 收到系统消息: %s。");
		format % _clientId % controller->getUserId() % argument->getRequest()->message();
		TestClientManager::getInstance()->getTestClient(_clientId)->addMessage(format.str());
		//Logger::getDefault()->log(format.str());
	}

	void ClientChatServiceImpl::onOtherPublicSay(OtherPublicSayMethod::Argument* argument)
	{
		auto controller = argument->getController();
		auto request = argument->getRequest();
		boost::format format("<客户端%d, 用户%d>: 收到用户%d公聊消息: %s。");
		format % _clientId % controller->getUserId() % request->other_user_id() % request->other_message();
		TestClientManager::getInstance()->getTestClient(_clientId)->addMessage(format.str());
		//Logger::getDefault()->log(format.str());
	}

	void ClientChatServiceImpl::onOtherPrivateSay(OtherPrivateSayMethod::Argument* argument)
	{
		auto controller = argument->getController();
		auto request = argument->getRequest();
		boost::format format("<客户端%d, 用户%d>: 收到用户%d私聊消息: %s。");
		format % _clientId % controller->getUserId() % request->other_user_id() % request->other_message();
		TestClientManager::getInstance()->getTestClient(_clientId)->addMessage(format.str());
		//Logger::getDefault()->log(format.str());
	}
}